#ifndef __MY_ASYNC_LOOPER_H__
#define __MY_ASYNC_LOOPER_H__

#include "AsyncBuffer.hpp"
#include <mutex>
#include <condition_variable>
#include <thread>
#include <functional>
#include <memory>
#include <atomic>

namespace LzcLog
{
    enum class AsyncType
    {
        ASYNC_SAFE, // 缓冲区固定长度，满了就阻塞，避免内除资源耗尽
        ASYNC_UNSAFE // 扩容，用于极限测试
    };
    using Func = std::function<void(Buffer&)>; // 不拷贝，直接使用Buffer里面的成员，所以不加const
    class AsyncLooper
    {
    public:
        using ptr = std::shared_ptr<AsyncLooper>;
        AsyncLooper(AsyncType async_type, const Func& callback)
        :_async_type(async_type)
        ,_running(true)
        ,_callback(callback)
        ,_thread([this](){this->ThreadEntry();})
        {}
        ~AsyncLooper(){
            Stop();
        }
        void Stop() {
            if(_running) {
                _running = false; // 为了线程安全要设置为原子类型
                _con_cond.notify_one(); // 唤醒仅有的一个消费者
                _thread.join();
            }
        }
        void Push(const char* data, size_t len) {
            if(_running) {
                {
                    std::unique_lock<std::mutex> lock (_mutex);
                    if(_async_type == AsyncType::ASYNC_SAFE) {
                        // 自动循环判断，生产者缓冲区空间是否足够。
                        _pro_cond.wait(lock, [this,len](){ 
                            return (this->_pro_buf).WriteAbleSize() >= len;
                        }); // 被唤醒+抢到锁+true，才往下走
                    }
                    _pro_buf.Push(data, len);
                }
                // (仅有的一个)消费者可能处理完消费者缓冲区数据，swap时，生产者缓冲区为空，在等待，需要被唤醒
                _con_cond.notify_one();
            }
        }
    private:
        void ThreadEntry() {
            // 可能在这个位置_running=false，但是生产者缓冲区还有数据，还要swap，进行处理
            while(_running || !_pro_buf.Empty())
            {
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    // 自动循环判断，生产者缓冲区空间 是否 非空 || 运行 是否 停止
                    // 能不能把!_running掉，不能，去掉后，当退出了，空，会继续等待生产者push，但是此时生产者不会再push了。
                    _con_cond.wait(lock, [this](){
                        return !_pro_buf.Empty() || !_running;
                    });
                    _con_buf.Swap(_pro_buf);
                }
                if(_async_type == AsyncType::ASYNC_SAFE && _running) // 只有运行，才唤醒
                    _pro_cond.notify_all();
                _callback(_con_buf); // 处理完数据
                _con_buf.Reset();
            }
        }
    private:
        AsyncType _async_type;
        std::atomic<bool> _running; // 异步工作器的停止标志
        Buffer _pro_buf; // 生产者缓冲区
        Buffer _con_buf; // 消费者缓冲区
        std::mutex _mutex;
        std::condition_variable _pro_cond;
        std::condition_variable _con_cond;
        std::thread _thread;
        Func _callback;
    };
}

#endif